package com.rickpan.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 分片上传队列管理器
 * 通过队列实现同一上传会话的串行处理，避免并发冲突
 */
@Component
public class ChunkUploadQueueManager {

    private static final Logger logger = LoggerFactory.getLogger(ChunkUploadQueueManager.class);

    // 每个uploadId对应一个锁，确保同一会话的分片串行处理
    private final ConcurrentHashMap<String, ReentrantLock> uploadLocks = new ConcurrentHashMap<>();
    
    // 线程池用于异步处理分片上传
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
        10, // 核心线程数
        50, // 最大线程数
        60L, TimeUnit.SECONDS, // 空闲线程存活时间
        new LinkedBlockingQueue<>(1000), // 队列容量
        r -> {
            Thread t = new Thread(r, "chunk-upload-" + System.currentTimeMillis());
            t.setDaemon(true);
            return t;
        }
    );

    /**
     * 获取上传会话的锁
     */
    private ReentrantLock getLock(String uploadId) {
        return uploadLocks.computeIfAbsent(uploadId, k -> new ReentrantLock());
    }

    /**
     * 串行执行分片上传任务
     */
    public <T> T executeSerially(String uploadId, UploadTask<T> task) throws Exception {
        ReentrantLock lock = getLock(uploadId);
        
        try {
            // 获取锁，确保同一uploadId的操作串行执行
            if (lock.tryLock(30, TimeUnit.SECONDS)) {
                try {
                    logger.debug("获取上传锁成功: uploadId={}", uploadId);
                    return task.execute();
                } finally {
                    lock.unlock();
                    logger.debug("释放上传锁: uploadId={}", uploadId);
                }
            } else {
                throw new RuntimeException("获取上传锁超时: " + uploadId);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("等待上传锁被中断: " + uploadId, e);
        }
    }

    /**
     * 异步串行执行分片上传任务
     */
    public void executeSeriallyAsync(String uploadId, Runnable task) {
        executor.submit(() -> {
            ReentrantLock lock = getLock(uploadId);
            
            try {
                if (lock.tryLock(30, TimeUnit.SECONDS)) {
                    try {
                        logger.debug("异步获取上传锁成功: uploadId={}", uploadId);
                        task.run();
                    } finally {
                        lock.unlock();
                        logger.debug("异步释放上传锁: uploadId={}", uploadId);
                    }
                } else {
                    logger.error("异步获取上传锁超时: uploadId={}", uploadId);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.error("异步等待上传锁被中断: uploadId={}", uploadId, e);
            } catch (Exception e) {
                logger.error("异步执行上传任务失败: uploadId={}", uploadId, e);
            }
        });
    }

    /**
     * 清理完成的上传会话锁
     */
    public void cleanupLock(String uploadId) {
        ReentrantLock lock = uploadLocks.remove(uploadId);
        if (lock != null) {
            logger.debug("清理上传锁: uploadId={}", uploadId);
        }
    }

    /**
     * 获取队列状态信息
     */
    public String getQueueStatus() {
        return String.format("活跃锁数: %d, 线程池状态: 活跃=%d, 队列=%d", 
                           uploadLocks.size(), 
                           executor.getActiveCount(), 
                           executor.getQueue().size());
    }

    /**
     * 上传任务接口
     */
    @FunctionalInterface
    public interface UploadTask<T> {
        T execute() throws Exception;
    }

    /**
     * 关闭线程池
     */
    public void shutdown() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
